package com.atguigu.mapreduce.reducejoin;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

public class ReduceJoinReducer extends Reducer<Text, OrderPd, OrderPd, NullWritable> {
    private ArrayList<OrderPd> pds = new ArrayList<>();
    private OrderPd pd = new OrderPd();
    @Override
    protected void reduce(Text key, Iterable<OrderPd> values, Reducer<Text, OrderPd, OrderPd, NullWritable>.Context context) throws IOException, InterruptedException {
        values.forEach(value -> {
            if (value.getTitle().equals("order.txt")){
                try {
                    OrderPd orderPd = new OrderPd();
                    BeanUtils.copyProperties(orderPd, value);
                    // 数据来源order.txt
                    pds.add(orderPd);
                } catch (IllegalAccessException e) {
                    throw new RuntimeException(e);
                } catch (InvocationTargetException e) {
                    throw new RuntimeException(e);
                }

            }else {
                // 数据来源pd.txt
                try {
                    BeanUtils.copyProperties(pd, value);
                } catch (IllegalAccessException e) {
                    throw new RuntimeException(e);
                } catch (InvocationTargetException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        pds.forEach(value -> {
            value.setpName(pd.getpName());
            try {
                context.write(value, NullWritable.get());
            } catch (IOException e) {
                throw new RuntimeException(e);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        pds.clear();
    }

}
